Lakehouse - Delta Lake & Best Practices: Hands-on Workbook

This workbook walks you through performing Delta upserts, time travel reads, and table compaction.

1. Delta Table Operations

• Task: Implement a safe upsert ETL pipeline using Delta Lake Merge mechanics and trace transaction logs.

2. Tasks

Task 1: Trace Merges & Time Travel

Write the PySpark Delta Lake code to:

1. Ingest a batch of customer updates.
2. Upsert (Merge) them into a target Delta table based on customer_id.
3. Query the table from a previous state using Time Travel.

Task 2: Implement Retention Purging via VACUUM

Write the PySpark/SQL expressions to purge unreferenced data files older than 48 hours using the VACUUM command, and detail the configuration updates required to bypass the default 7-day safety check.

3. Step-by-Step Solutions

Solution 1: Delta code implementation

• PySpark Code:

from delta.tables import DeltaTable

# 1. Read updates
updates_df = spark.read.parquet("/tmp/updates")

# 2. Reference the active Delta Table
deltaTable = DeltaTable.forPath(spark, "/mnt/delta/customers")

# 3. Perform atomic Merge (Upsert)
deltaTable.alias("target").merge(
    updates_df.alias("updates"),
    "target.customer_id = updates.customer_id"
).whenMatchedUpdate(set = {
    "email": "updates.email",
    "last_purchase_date": "updates.last_purchase_date"
}).whenNotMatchedInsert(values = {
    "customer_id": "updates.customer_id",
    "email": "updates.email",
    "last_purchase_date": "updates.last_purchase_date"
}).execute()

# 4. Read Table at a previous state (Time Travel Version 2)
df_v2 = spark.read.format("delta").option("versionAsOf", 2).load("/mnt/delta/customers")

# 5. Compact the Table files using Optimize
spark.sql("OPTIMIZE '/mnt/delta/customers' Z-ORDER BY (last_purchase_date)")

Solution 2: Vacuum & Retention Configuration

• Bypassing Retention Checks: By default, Delta blocks running vacuum with values below 168 hours (7 days) to prevent active writers from crashing. We must disable this check in the Spark session configuration:

# 1. Disable safety check
spark.conf.set("spark.databricks.delta.vacuum.parallelDelete.enabled", "true")
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")

# 2. Execute Vacuum with 48 hours retention
spark.sql("VACUUM '/mnt/delta/customers' RETAIN 48 HOURS")